package com.dy.dws.UserPatientAddWin;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSONObject;
import com.dy.dws.UserRegisterWin.DwsUserRegisterWin;
import com.dy.realtime.common.base.BaseApp;
import com.dy.realtime.common.bean.DwsUserPatientAddBean;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.DateFormatUtil;
import com.dy.realtime.common.util.DorisMapFunction;
import com.dy.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.checkerframework.checker.units.qual.C;

import java.time.Duration;
import java.util.ArrayList;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.dws.UserPatientAddWin
 * @className: DwsUserPatientAddWin
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/16 18:35
 * @version: 1.0
 */

public class DwsUserPatientAddWin extends BaseApp {
    public static void main(String[] args) {
        new DwsUserPatientAddWin().start(Constant.TOPIC_DWD_TRADE_USER_PATIENT_ADD, Constant.TOPIC_DWS_TRADE_USER_PATIENT_ADD_WIN, 4, 10026);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        SingleOutputStreamOperator<DwsUserPatientAddBean> etlStream = getEtlStream(dataStreamSource);
//        etlStream.print();
        SingleOutputStreamOperator<DwsUserPatientAddBean> result = getProcess(etlStream);
//        result.print();
         result.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_user_patient_add_win"));
    }

    private static SingleOutputStreamOperator<DwsUserPatientAddBean> getProcess(SingleOutputStreamOperator<DwsUserPatientAddBean> etlStream) {
        return etlStream.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsUserPatientAddBean>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<DwsUserPatientAddBean>() {
                    @Override
                    public long extractTimestamp(DwsUserPatientAddBean dwsUserPatientAddBean, long l) {
                        return dwsUserPatientAddBean.getTs();
                    }
                })).windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessAllWindowFunction<DwsUserPatientAddBean, DwsUserPatientAddBean, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<DwsUserPatientAddBean, DwsUserPatientAddBean, TimeWindow>.Context context, Iterable<DwsUserPatientAddBean> iterable, Collector<DwsUserPatientAddBean> collector) throws Exception {
                        ArrayList<DwsUserPatientAddBean> dwsUserPatientAddBeans = ListUtil.toList(iterable);
                        long count = dwsUserPatientAddBeans.stream().map(e -> e.getId()).count();
                        String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDate = DateFormatUtil.tsToDate(System.currentTimeMillis());
                        DwsUserPatientAddBean next = dwsUserPatientAddBeans.iterator().next();
                        collector.collect(DwsUserPatientAddBean.builder()
                                .id(next.getId())
                                .userId(next.getUserId())
                                .birthday(next.getBirthday())
                                .createTime(next.getCreateTime())
                                .curDate(curDate)
                                .edt(edt)
                                .stt(stt)
                                .patientAddCount(count)
                                .name(next.getName())
                                .gender(next.getGender())
                                .build());
                    }
                });
    }

    private static SingleOutputStreamOperator<DwsUserPatientAddBean> getEtlStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, DwsUserPatientAddBean>() {
            @Override
            public void flatMap(String s, Collector<DwsUserPatientAddBean> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                String create_time = jsonObject.getString("create_time");
                if (create_time != null) {
                    DwsUserPatientAddBean dwsUserPatientAddBean = JSONObject.parseObject(jsonObject.toString(), DwsUserPatientAddBean.class);
                    dwsUserPatientAddBean.setPatientAddCount(1L);
                    collector.collect(dwsUserPatientAddBean);
                }
            }
        });
    }
}
